package com.houger.ddd.redpacket.infra.mq.kafka;

import com.alibaba.fastjson.JSON;
import com.houger.ddd.redpacket.domain.GrabRedPacketDomain;
import com.houger.ddd.redpacket.facade.RedPacketFacade;
import jakarta.annotation.Resource;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

// 消费者
@Component
public class Consumer {
	@Resource
	RedPacketFacade redPacketFacade;

	@KafkaListener(topics = { "rd_grab" },groupId = "ddd",containerFactory = "kafkaListenerContainerFactory")
	public void receiveMessage(String message, Acknowledgment ack) {
		final GrabRedPacketDomain domain = JSON.parseObject(message, GrabRedPacketDomain.class);
		// 异步保持到db
		redPacketFacade.saveGrabToDb(domain);
		redPacketFacade.updateResidual(domain.getRedPacketId(),domain.getMoney());
		ack.acknowledge();//直接提交offset
	}
}

